/*
 * Copyright 2007 Stefan Arentz <http://stefan.arentz.nl>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.sateh.spaces.simple;

import com.sateh.spaces.Lease;
import com.sateh.spaces.SpaceConnection;
import com.sateh.spaces.SpaceException;

import javax.transaction.Transaction;
import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

/**
 * This class implements the proxy for the @{link SimpleSpace} space implementation. It is Stefan's
 * first naive try to implement his own flavour of the JavaSpaces specification.
 *
 * @author Stefan Arentz <stefan@arentz.nl>
 */

public class SimpleSpaceConnection implements SpaceConnection
{
    /**
     *
     */
    
    private final SimpleSpace space;

    /**
     * 
     */

    private final static Map<String,SimpleSpace> spaces = new HashMap<String,SimpleSpace>();

    /**
     *
     */

    private final String name;

    //

    SimpleSpaceConnection(String name)
    {
        if (name == null) {
            throw new IllegalArgumentException("Argument 'name' cannot be null");
        }

        if (name.length() == 0) {
            throw new IllegalArgumentException("Invalid argument 'spaceUrl'");
        }

        this.name = name;

        synchronized(spaces) {
            SimpleSpace simpleSpace = spaces.get(name);
            if (simpleSpace == null) {
                simpleSpace = new SimpleSpace();
                spaces.put(name, simpleSpace);
            }
            this.space = simpleSpace;
        }
    }

    //

    public SimpleSpace getSpace()
    {
        return space;
    }

    public String getName()
    {
        return name;
    }

    //

    /**
     * @see com.sateh.spaces.Space
     */

    public <T> T read(Class<T> objectClass, Transaction tx, long timeout)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().readOrWaitForSpaceObject(objectClass.getName(), timeout);
        if (spaceObject != null)
        {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space
     */

    public <T> T readIfExists(Class<T> objectClass, Transaction tx)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().readSpaceObject(objectClass.getName());
        if (spaceObject != null)
        {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space
     */

    public <T> T read(T templateObject, Transaction tx, long timeout) throws SpaceException
    {
        Object object = null;

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);
        
        SimpleSpaceObject spaceObject = getSpace().readOrWaitForSpaceObject(template, timeout);
        if (spaceObject != null)
        {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space
     */

    public <T> T readIfExists(T templateObject, Transaction tx) throws SpaceException
    {
        Object object = null;

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);

        SimpleSpaceObject spaceObject = getSpace().readSpaceObject(templateObject.getClass().getName(), template);
        if (spaceObject != null)
        {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    //

    /**
     * @see com.sateh.spaces.Space
     */

    public Lease write(Object object, Transaction tx, long timeToLive)
        throws SpaceException
    {
        if (object == null) {
            throw new IllegalArgumentException("Argument 'object' cannot be null");
        }

        if (!(object instanceof Serializable)) {
            throw new IllegalArgumentException("Argument 'object' must implement Serializable");
        }

        if (timeToLive < 0) {
            throw new IllegalArgumentException("Argument 'timeToLive' cannot be negative");
        }

        try
        {
            SimpleSpaceObject spaceObject = new SimpleSpaceObject(object, timeToLive);
            
            return new SimpleSpaceLease(
                getSpace(),
                getSpace().writeSpaceObject(spaceObject, timeToLive),
                timeToLive
            );
        }

        catch (IOException exception)
        {
            throw new SpaceException("Cannot serialize the object: " + exception.getMessage(), exception);
        }
    }

    //

    /**
     * @see com.sateh.spaces.Space
     */

    public <T> T take(Class<T> objectClass, Transaction tx, long timeout)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().takeOrWaitForSpaceObject(objectClass.getName(), timeout);
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space
     */

    public <T> T takeIfExists(Class<T> objectClass, Transaction tx)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().takeSpaceObject(objectClass.getName());
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#take(Object, javax.transaction.Transaction, long) 
     */

    public <T> T take(T templateObject, Transaction tx, long timeout)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);

        SimpleSpaceObject spaceObject = getSpace().takeOrWaitForSpaceObject(templateObject.getClass().getName(), template, timeout);
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#takeIfExists(Object, javax.transaction.Transaction) 
     */

    public <T> T takeIfExists(T templateObject, Transaction tx)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);

        SimpleSpaceObject spaceObject = getSpace().takeSpaceObject(templateObject.getClass().getName(), template);
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    // Count

    /**
     * @see com.sateh.spaces.Space#count(Class)
     */

    public long count(Class<? extends Serializable> clazz) throws SpaceException
    {
        if (clazz == null) {
            throw new IllegalArgumentException("Argument 'clazz' cannot be null");
        }

        return getSpace().count(clazz.getName());
    }

    /**
     * @see com.sateh.spaces.Space#count(java.io.Serializable)  
     */

    public long count(Serializable templateObject) throws SpaceException
    {
        if (templateObject == null) {
            throw new IllegalArgumentException("Argument 'templateObject' cannot be null");
        }

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);
        
        return getSpace().count(template);
    }
}
